use paho_mqtt::{CreateOptionsBuilder, AsyncClient, ServerResponse, Message, ConnectOptions};
use std::time::Duration;
use tokio::task;
use futures_util::StreamExt;
use tokio::sync::mpsc:: Sender;
use crate::common::cs;
use log::error;
use crate::mqtt::{MqttVersion, MqttQos, MQTT_VERSION_DEFAULT, QOS_2};


/// Mqtt客户端集成
#[derive(Clone)]
pub struct MqttClient {
    server_uri: String,
    client_id: String,
    client: AsyncClient,
    mqtt_version: MqttVersion,
    publish_qos: MqttQos,
    subscribe_qos: MqttQos,
}

impl MqttClient {
    /// 创建异步mqtt客户端
    pub fn new(server_host: &str, server_port: u16, client_id: &str) -> cs::Result<Self>
    {
        let server_uri = format!("tcp://{}:{}", server_host, server_port);
        let client = CreateOptionsBuilder::new()
            .server_uri(&server_uri)
            .client_id(client_id)
            .mqtt_version(MQTT_VERSION_DEFAULT)
            .create_client()?;
        Ok(Self {
            server_uri,
            client_id:
            client_id.to_string(),
            client,
            mqtt_version: MQTT_VERSION_DEFAULT,
            publish_qos: QOS_2,
            subscribe_qos: QOS_2,
        })
    }

    /// 客户端链接
    pub async fn connect(&self, option: ConnectOptions) -> cs::Result<ServerResponse>
    { Ok(self.client.connect(option).await?) }

    /// 默认客户端链接
    pub async fn connect_default(&self) -> cs::Result<ServerResponse>
    {
        let option = paho_mqtt::ConnectOptionsBuilder::new()
            .keep_alive_interval(Duration::from_secs(60))
            .clean_session(false)
            .automatic_reconnect(
                Duration::from_secs(5),
                Duration::from_secs(30),
            )
            .finalize();
        Ok(self.client.connect(option).await?)
    }

    /// 订阅主题并监听
    pub async fn subscribe(&mut self, topic: &str, sender: Sender<Message>) -> cs::Result<()>
    {
        let resp = self.client.subscribe(topic, self.subscribe_qos).await?;
        println!(" - mqtt subscribe `{}` : `{}`", topic, resp.reason_code());
        let mut receiver = self.client.get_stream(64);
        task::spawn(async move {
            loop {
                match receiver.next().await {
                    Some(Some(msg)) => {
                        if let Err(e) = sender.send(msg).await {
                            error!("{}", e);
                            break;
                        }
                    }
                    _ => { continue; }
                }
            }
        });
        Ok(())
    }

    /// 发送消息
    pub async fn publish<T, P>(&self, topic: T, payload: P) -> cs::Result<()>
        where T: Into<String>,
              P: Into<Vec<u8>>
    {
        let message = Message::new(topic, payload, self.publish_qos);
        Ok(self.client.publish(message).await?)
    }
}

/// GET&SET
impl MqttClient {
    pub fn set_server_uri(&mut self, server_uri: String) {
        self.server_uri = server_uri;
    }
    pub fn set_client_id(&mut self, client_id: String) {
        self.client_id = client_id;
    }
    pub fn set_client(&mut self, client: AsyncClient) {
        self.client = client;
    }
    pub fn set_mqtt_version(&mut self, mqtt_version: MqttVersion) {
        self.mqtt_version = mqtt_version;
    }
    pub fn set_publish_qos(&mut self, publish_qos: MqttQos) {
        self.publish_qos = publish_qos;
    }
    pub fn set_subscribe_qos(&mut self, subscribe_qos: MqttQos) {
        self.subscribe_qos = subscribe_qos;
    }
    pub fn server_uri(&self) -> &str {
        &self.server_uri
    }
    pub fn client_id(&self) -> &str {
        &self.client_id
    }
    pub fn client(&self) -> &AsyncClient {
        &self.client
    }
    pub fn mqtt_version(&self) -> MqttVersion {
        self.mqtt_version
    }
    pub fn publish_qos(&self) -> MqttQos {
        self.publish_qos
    }
    pub fn subscribe_qos(&self) -> MqttQos {
        self.subscribe_qos
    }
}
